package com.zhouqinuan.spring.cloud.develop.framework.rabbitmq.consumer;

import java.io.IOException;
import java.util.Date;

import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class AbstractMessageConsumer<T> implements MessageConsumer<T> {

    @Autowired
    private ObjectMapper objectMapper;
    
    @Override
    public void processMessage(T entity, Channel channel, Message message) {
        
        if (log.isDebugEnabled()) {
            try {
                log.debug("process message: {}", objectMapper.writeValueAsString(entity));
            } catch (Exception e) {}
        }
        
        Date nowTime = new Date();
        boolean isAck = false;
        int maxRetryTime = getMaxRetryTime();
        int delayTime = message.getMessageProperties().getReceivedDelay() == null ? 0 : message.getMessageProperties().getReceivedDelay().intValue();
        
        try {
            
            onProcessMessage(entity, message);
            
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            
            isAck = true;
            
        } catch (Exception e) {
            
            if (message.getMessageProperties().getTimestamp() != null && nowTime.getTime() - message.getMessageProperties().getTimestamp().getTime() <= (maxRetryTime + delayTime)) {
                try {
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                } catch (IOException ioException) {
                    log.error("basicNack error: {}", ioException);
                }
            }
            
        } finally {
            
            if (message.getMessageProperties().getTimestamp() == null || (!isAck && nowTime.getTime() - message.getMessageProperties().getTimestamp().getTime() > (maxRetryTime + delayTime))) {
                
                try {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    log.error("basicAck error: {}", e);
                }
            }
        }
    }

    /*实际处理消息的逻辑*/
    protected abstract void onProcessMessage(T entity, Message message) throws Exception;

    /*最大重试时间，单位毫秒，小心这个参数，可能会导致消息堆积*/
    protected abstract int getMaxRetryTime();
}
